package com.yiguo.realtime.app

import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * 四个需求都需要使用SparkSteaming从kafka消费数据，因此开发流程是一样的
 *
 *      1、创建一个 Steamingcontext
 *      2、从kafka获取ds
 *      3、对Ds进行准换
 *            四个需要对ds进行不同的转换
 *      4、启动app
 *      5、阻塞当前线程
 */
abstract class BaseApp {
      var appName:String
      var duration:Int
  //运行程序
  var streamingContext:StreamingContext = null
  def run(op: =>Unit): Unit ={
    try{
      //先创建StreamingContext
      streamingContext = new StreamingContext("local[*]", appName, Seconds(duration))
      //具体操作
        op
      //启动
      streamingContext.start()
      //阻塞线程
      streamingContext.awaitTermination()
    }catch {
      case ex:Exception => ex.printStackTrace()
        streamingContext.stop(true)
    }
  }

}
